package com.harzone.commons.search.impl.service;

import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.http.client.HttpClient;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrQuery.SortClause;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpClientUtil;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.params.CursorMarkParams;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.harzone.commons.search.domain.AFunctions;
import com.harzone.commons.search.domain.Field;
import com.harzone.commons.search.domain.Record;
import com.harzone.commons.search.exception.SearchException;
import com.harzone.commons.search.impl.maker.SearchResultMaker;
import com.harzone.commons.search.impl.mapper.ScanRowMapperAdapter;
import com.harzone.commons.search.impl.statement.DeleteParser;
import com.harzone.commons.search.impl.statement.SelectParser;
import com.harzone.commons.search.result.RowMapper;
import com.harzone.commons.search.result.ScanRowMapper;
import com.harzone.commons.search.result.SearchResult;
import com.harzone.commons.search.result.SearchResultSet;
import com.harzone.commons.search.result.SearchResultSetImpl;
import com.harzone.commons.search.service.SearchService;
import com.harzone.commons.search.statement.Delete;
import com.harzone.commons.search.statement.Insert;
import com.harzone.commons.search.statement.Select;
import com.harzone.commons.search.statement.Update;

/**
 * 搜索服务solr实现
 * 
 * @author ls 2014年11月21日下午3:40:25
 */
public class SearchServiceImpl implements SearchService {

	protected static final String DEFAULT_COLLECTION = "collection1";
	protected static final int DEFAULT_COMMIT_MS = 10;

	protected Logger logger = LoggerFactory.getLogger(this.getClass());
	// solr服务
	private SolrClient server = null;
	// 服务地址
	private String address = "";
	// 服务是否关闭
	private boolean closed = true;
	// 是否检查是否存在
	private boolean checkExist = false;
	// 是否自动提交
	private boolean autoCommit = false;
	// 主键名称
	private String primaryFieldName = "id";

	public SearchServiceImpl() {
		super();
	}

	public SearchServiceImpl(String address) {
		this(address, false, false);
	}

	/**
	 * 
	 * @param address
	 *            地址
	 * @param checkExist
	 *            是否检查是否存在
	 * @param autoCommit
	 *            是否自动提交
	 */
	public SearchServiceImpl(String address, boolean checkExist,
			boolean autoCommit) {
		super();
		this.address = address;
		this.checkExist = checkExist;
		this.autoCommit = autoCommit;
	}

	public static SearchService newInstance(String address) {
		return new SearchServiceImpl(address);
	}

	@Override
	public boolean open(String address) {
		// 是否为solrcloud
		boolean isCloud = false;
		if (!address.startsWith("http:")) {
			isCloud = true;
		}
		this.close();
		if (isCloud) {
			logger.info("...准备打开solrcloud服务【" + address + "】...");
			this.server = this.getCloudSolrServer(address, DEFAULT_COLLECTION);
			logger.info("...已打开solrcloud服务...");
		} else {
			logger.info("...准备打开solr服务【" + address + "】...");
			this.server = new HttpSolrClient(address);
			logger.info("...已打开solr服务...");
		}
		this.closed = false;
		this.address = address;
		return false;
	}

	/**
	 * 获取solrcloud
	 * 
	 * @author ls 2014年11月21日下午3:34:12
	 * @param zkHost
	 * @param defaultCollection
	 * @return
	 */
	protected CloudSolrClient getCloudSolrServer(String zkHost,
			String defaultCollection) {
		synchronized (this) {
			if (server == null) {
				ModifiableSolrParams params = new ModifiableSolrParams();
				params.set(HttpClientUtil.PROP_MAX_CONNECTIONS, 1000);// 10
				params.set(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, 500);// 5
				HttpClient client = HttpClientUtil.createClient(params);
				LBHttpSolrClient lbServer = new LBHttpSolrClient(client);

				CloudSolrClient server1 = null;
				server1 = new CloudSolrClient(zkHost, lbServer, true);
				server1.setDefaultCollection(defaultCollection);
				server1.setZkClientTimeout(30000);
				server1.setZkConnectTimeout(30000);
				if (defaultCollection != null) {
					server1.setDefaultCollection(defaultCollection);
				}
				server1.setParallelUpdates(true);
				server1.setDefaultCollection(DEFAULT_COLLECTION);
				server1.connect();
				this.server = server1;
			}
		}
		return (CloudSolrClient) server;
	}

	@Override
	public boolean open() {
		return this.open(this.address);
	}

	@Override
	public boolean reopen() {
		if (this.closed || this.server == null) {
			this.open();
		} else {
			this.close();
			this.open();
		}
		return false;
	}
	
	@Override
	public boolean commit() {
		return this.commit(false);
	}

	@Override
	public boolean commit(boolean softCommit) {
		boolean flag = false;
		logger.info("提交请求...");
		long start = System.currentTimeMillis();
		try {
			if(softCommit){
				this.server.commit(false, false, true);
			}else{
				this.server.commit();
			}
			flag = true;
		} catch (SolrServerException e) {
			throw new SearchException(e.getMessage(), e);
		} catch (IOException e) {
			throw new SearchException(e.getMessage(), e);
		}
		this.logger.info("提交请求操作耗时：" + (System.currentTimeMillis() - start)
				+ " ms");
		return flag;
	}

	@Override
	public boolean rollback() {
		boolean flag = false;
		logger.info("回滚请求...");
		long start = System.currentTimeMillis();
		try {
			this.server.rollback();
			flag = true;
		} catch (SolrServerException e) {
			throw new SearchException(e.getMessage(), e);
		} catch (IOException e) {
			throw new SearchException(e.getMessage(), e);
		}
		this.logger.info("回滚请求操作耗时：" + (System.currentTimeMillis() - start)
				+ " ms");
		return flag;
	}

	@Override
	public boolean insert(Insert insert) {
		return this.insert(insert, true);
	}

	@Override
	public boolean insert(Insert insert, boolean checkExist) {
		boolean flag = false;
		if (insert == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		List<Record> list = insert.getRecords();
		if (list != null && !list.isEmpty()) {
			Collection<SolrInputDocument> docs = new ArrayList<SolrInputDocument>(
					list.size());
			for (Record record : list) {
				SolrInputDocument doc = new SolrInputDocument();
				String id = "";
				List<Field> fields = record.getFields();
				for (Field field : fields) {
					doc.addField(field.getIndexName(), field.getValue(),
							field.getBoost());
					if (field.getIndexName().equals(this.primaryFieldName)) {
						id = String.valueOf(field.getValue());
					}
				}
				if (checkExist && this.isExist(id)) {
					throw new SearchException("唯一键值【" + id
							+ "】已经存在！本次添加的数据添加失败!");
				}
				docs.add(doc);
			}
			flag = this.add(docs);
		}
		this.logger.info("添加操作耗时：" + (System.currentTimeMillis() - start)
				+ " ms");
		return flag;
	}

	/**
	 * 该主键值是否存在
	 * 
	 * @author ls 2014年11月21日下午5:35:48
	 * @param priKey
	 * @return
	 */
	private boolean isExist(String priKey) {
		boolean flag = false;
		SolrQuery query = new SolrQuery("*:*");
		query.addFilterQuery(this.primaryFieldName + ":" + priKey);
		query.setFacet(true);
		query.setStart(0);
		query.setRows(0);
		try {
			QueryResponse response = server.query(query);
			long count = response.getResults().getNumFound();
			if (count > 0) {
				flag = true;
			}
		} catch (Exception e) {
			flag = false;
			throw new SearchException(e.getMessage(), e);
		}
		return flag;
	}

	@Override
	public boolean delete(Delete delete) {
		if (delete == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		boolean flag = false;
		DeleteParser parser = new DeleteParser(delete);
		SolrQuery query = parser.getQuery();
		boolean hasCondition = parser.hasCondition();
		try {
			if (delete.hasId()) {
				this.logger.info("删除：" + delete.getIds());
				this.server.deleteById(delete.getIds(), DEFAULT_COMMIT_MS);
			}
			if (!hasCondition) {
				this.logger.info(query.getQuery());
				this.server.deleteByQuery(query.getQuery(), DEFAULT_COMMIT_MS);
			}
			flag = true;
		} catch (SolrServerException e) {
			throw new SearchException(e.getMessage(), e);
		} catch (IOException e) {
			throw new SearchException(e.getMessage(), e);
		} finally {
			if (this.autoCommit) {
				if (flag) {
					this.commit();
				} else {
					this.rollback();
				}
			}
		}
		this.logger.info("删除操作耗时：" + (System.currentTimeMillis() - start)
				+ " ms");
		return flag;
	}

	@Override
	public boolean update(Update update) {
		return this.update(update, false);
	}

	@Override
	public boolean update(Update update, boolean checkExist) {
		if (update == null) {
			return false;
		}
		boolean flag = false;
		long start = System.currentTimeMillis();
		List<Record> list = update.getRecords();
		final Collection<SolrInputDocument> docs = new ArrayList<SolrInputDocument>();
		if (update.hasWhere()) {
			if (list == null || list.isEmpty()) {
				return false;
			}
			Select select = new Select();
			select.setWhere(update.getWhere());
			final Record updateRecord = list.get(0);
			try {
				this.scan(select, new ScanRowMapperAdapter() {
					
					@Override
					public boolean processingOfResults(List<Record> list, int page) {
						if(list!=null){
							 for (Record record : list) {
								    SolrInputDocument doc = new SolrInputDocument();
									List<Field> fields = record.getFields();
									for (Field field : fields) {
										doc.addField(field.getIndexName(), field.getValue(),
												field.getBoost());
									}
									fields = updateRecord.getFields();
									for (Field field : fields) {
//										if(doc.containsKey(field.getIndexName())){
//											doc.remove(field.getIndexName());
//										}
										doc.setField(field.getIndexName(), field.getValue(),
												field.getBoost());
									}
									docs.add(doc);
							 }
						}
						return true;
					}
				});
			} catch (SQLException e) {
				throw new SearchException(e.getMessage(), e);
			}
		} else {
			if (list != null && !list.isEmpty()) {
				for (Record record : list) {
					SolrInputDocument doc = new SolrInputDocument();
					String id = "";
					List<Field> fields = record.getFields();
					for (Field field : fields) {
						doc.addField(field.getIndexName(), field.getValue(),
								field.getBoost());
						if (field.getIndexName().equals(this.primaryFieldName)) {
							id = String.valueOf(field.getValue());
						}
					}
					if (checkExist && !this.isExist(id)) {
						throw new SearchException("唯一键值【" + id + "】不存在！更新失败!");
					}
					docs.add(doc);
				}
			}
		}
		flag = this.add(docs); 
		this.logger.info("更新操作耗时：" + (System.currentTimeMillis() - start)
				+ " ms");
		return flag;
	}
	
	
	/**
	 * 添加到服务
	 * @author ls
	 * 2014年11月25日下午7:46:24 
	 * @param docs
	 * @return
	 */
	private boolean add(final Collection<SolrInputDocument> docs){
		boolean flag = false;
		try {
			if (this.logger.isDebugEnabled()) {
				this.logger.debug(docs.toString());
			}
			this.server.add(docs);
			flag = true;
		} catch (SolrServerException e) {
			throw new SearchException(e.getMessage(), e);
		} catch (IOException e) {
			throw new SearchException(e.getMessage(), e);
		} finally {
			if (this.autoCommit) {
				if (flag) {
					this.commit();
				} else {
					this.rollback();
				}
			}
		}
		return flag;
	}

	@Override
	public SearchResult search(Select select) {
		if (select == null) {
			return new SearchResult();
		}
		long start = System.currentTimeMillis();
		SelectParser parser = new SelectParser(select);
		SolrQuery query = parser.getQuery();
		QueryResponse response = null;
		this.logger.info(query.toString());
		try {
			response = this.server.query(query);
		} catch (Exception e) {
			throw new SearchException(e.getMessage(), e);
		}
		SearchResult result = null;
		if (select.isGroup) {
			result = SearchResultMaker.makeGroupResult(
					response.getGroupResponse(), select);
		} else {
			result = SearchResultMaker
					.makeResult(response.getResults(), select);
		}
		this.logger.info("查询操作耗时：" + (System.currentTimeMillis() - start)
				+ " ms");
		return result;
	}

	@Override
	public void close() {
		if (this.server != null) {
			logger.info("关闭solr服务...");
			try {
				this.server.close();
			} catch (Exception e) {
				e.printStackTrace();
			}
			server = null;
		}
		this.closed = true;
	}

	@Override
	public boolean isClosed() {
		return this.closed;
	}

	@Override
	public String getAddress() {
		return this.address;
	}

	public void setClosed(boolean closed) {
		this.closed = closed;
	}

	public boolean isCheckExist() {
		return checkExist;
	}

	public void setCheckExist(boolean checkExist) {
		this.checkExist = checkExist;
	}

	@Override
	public boolean isAutoCommit() {
		return autoCommit;
	}

	public void setAutoCommit(boolean autoCommit) {
		this.autoCommit = autoCommit;
	}

	public String getPrimaryFieldName() {
		return primaryFieldName;
	}

	public void setPrimaryFieldName(String primaryFieldName) {
		this.primaryFieldName = primaryFieldName;
	}

	@Override
	public boolean execute(String sql) {
		return this.execute(sql, false);
	}

	@Override
	public boolean execute(String sql, boolean checkExist) {
		// TODO Auto-generated method stub
		short type = 0;// 0:add,1:update,2:delete
		if (type == 0) {
			Insert insert = null;
			return this.insert(insert, checkExist);
		} else if (type == 1) {
			Update update = null;
			return this.update(update);
		} else {
			Delete delete = null;
			return this.delete(delete);
		}
	}

	@Override
	public int executeQueryForInt(String sql) throws SQLException {
		// TODO Auto-generated method stub
		Select select = null;
		return this.executeQueryForInt(select);
	}

	@Override
	public SearchResult executeQuery(String sql) {
		// TODO Auto-generated method stub
		Select select = null;
		return this.search(select);
	}

	@Override
	public SearchResultSet executeQuery(Select select) {
		SearchResult result = this.search(select);
		return new SearchResultSetImpl(result);
	}

	@Override
	public<T> List<T> executeQuery(Select select, final RowMapper<T> rowMapper)
			throws SQLException {
		SearchResultSet result = this.executeQuery(select);
		List<T> list = new ArrayList<T>(result.getFetchSize());
		while (result.next()) {
			T obj = rowMapper.mapRow(result);
			if (obj != null) {
				list.add(obj);
			}
		}
		result.close();
		return list;
	}

	@Override
	public int executeQueryForInt(Select select) throws SQLException {
		int count = 0;
		SearchResultSet result = this.executeQuery(select);
		int line = 0;
		int index = 0;
		while (result.next()) {
			count += result.getInt(AFunctions.COUNT.getName());
			line++;
			index++;
			if (index == 2) {
				line = index;
			} else if (index == 1) {
				if (count == 0) {
					line = 0;
				}
			}
			// count = result.getInt(0);
		}
		return Math.max(count, line);
	}

	/**
	 * 使用游标
	 * @author ls
	 * 2014年11月25日下午7:19:59 
	 * @param query
	 * @param rowCount
	 * @param rowMapper
	 * @throws SolrServerException
	 * @throws SQLException
	 * @throws IOException 
	 */
	private <T> void cursor(SolrQuery query, int rowCount, ScanRowMapper<T> rowMapper)
			throws SolrServerException, SQLException, IOException {
		SolrQuery q = query.setRows(rowCount);
		List<SortClause> list = q.getSorts();
		boolean exist = false;
		if(list != null){
			for (SortClause sortClause : list) {
				if(sortClause.getItem().equals(this.primaryFieldName)){
					exist = true;
					break;
				}
			}
		}
		if(!exist){
			q.addSort(SortClause.desc(this.primaryFieldName));
		}
		String cursorMark = CursorMarkParams.CURSOR_MARK_START;
		boolean done = false;
		int page = 1;
		boolean showField = true;
		long count = 0, showCount = 0;
		String fl = query.get("fl");
		while (!done) {
			showField = rowMapper.showField(page);
			if (!showField) {
				q.set("fl", this.primaryFieldName);
			}else{
				q.set("fl", fl);
			}
			long start = System.currentTimeMillis();
			q.set(CursorMarkParams.CURSOR_MARK_PARAM, cursorMark);
			this.logger.info(q.toString());
			QueryResponse rsp = server.query(q);
			count = rsp.getResults().getNumFound();
			this.logger.info("本次遍历" + (page) + "操作耗时："
					+ (System.currentTimeMillis() - start) + " ms");
			boolean goon = this.processingOfResults(rsp, showField, page,
					rowMapper);
			String nextCursorMark = rsp.getNextCursorMark();
			showCount = rowCount * page;
			if (showCount == count || !goon
					|| cursorMark.equals(nextCursorMark)) {
				done = true;
			}
			++page;
			cursorMark = nextCursorMark;
		}
	}

	
	/**
	 * 处理结果
	 * @author ls
	 * 2014年11月25日下午7:19:49 
	 * @param rsp
	 * @param showField
	 * @param page
	 * @param rowMapper
	 * @return
	 * @throws SQLException
	 */
	private <T> boolean processingOfResults(QueryResponse rsp, boolean showField,
			int page, ScanRowMapper<T> rowMapper) throws SQLException {
		List<T> list = null;
		if (showField) {
			SearchResult result = SearchResultMaker.makeResult(
					rsp.getResults(), null);
			SearchResultSet resultSet = new SearchResultSetImpl(result);
			list = new ArrayList<T>(resultSet.getFetchSize());
			while (resultSet.next()) {
				T obj = rowMapper.mapRow(resultSet);
				if (obj != null) {
					list.add(obj);
				}
			}
			resultSet.close();
		}
		boolean goon = rowMapper.processingOfResults(list, page);
		return goon;
	}

	@Override
	public <T> void scan(Select select, ScanRowMapper<T> rowMapper)
			throws SQLException {
		SelectParser parser = new SelectParser(select);
		SolrQuery query = parser.getQuery();
		try {
			this.cursor(query, select.getRowCount(), rowMapper);
		} catch (Exception e) {
			throw new SQLException(e.getMessage(), e);
		}
	}

}
